package com.hyw.trace.configuration.meta.redis.queue;

import java.util.Arrays;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.Nullable;

class IdempotentRedisQueue extends AbstractCacheQueue {

  private static final Logger log = LoggerFactory.getLogger(IdempotentRedisQueue.class);

  private static final String IdempotentPrefix = "idempotent:queue:";

  private Function<String, Entry<String, String>> keyExtractor;

  private String idempotentTopic;

  IdempotentRedisQueue(Function<String, Entry<String, String>> keyExtractor, String topic,
      RedisTemplate<String, String> redisTemplate, @Nullable Consumer<String> consumer) {
    super(topic, redisTemplate, consumer);
    Objects.requireNonNull(keyExtractor, "keyExtractor不允许为空");
    this.keyExtractor = keyExtractor;
    this.idempotentTopic = IdempotentPrefix + topic;
  }

  @Override
  protected void run() {
    while (isRunning()) {
      String msg = redisTemplate.opsForList().rightPop(getTopic(), timeout.getSeconds(), TimeUnit.SECONDS);
      if (Objects.nonNull(msg)) {

        // 同时维护消息去重
        String fieldKey = this.keyExtractor.apply(msg).getKey();
        redisTemplate.<String, String> opsForHash().delete(idempotentTopic, fieldKey);

        try {
          this.handle(msg);
        } catch (Exception ex) {
          log.error("", ex);
        }
      }
    }
  }

  @Override
  public boolean add(String msg) {
    super.add(msg);
    boolean modify = true;
    Entry<String, String> entry = this.keyExtractor.apply(msg);
    if (Objects.equals(entry.getValue(),
        this.redisTemplate.<String, String> opsForHash().get(idempotentTopic, entry.getKey()))) {
      modify = false;
    } else {
      redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
        connection.hSet(this.idempotentTopic.getBytes(), entry.getKey().getBytes(), entry.getValue().getBytes());
        connection.lPush(this.getTopic().getBytes(), msg.getBytes());
        return null;
      });
    }
    return modify;
  }

  @Override
  public void clear() {
    redisTemplate.delete(Arrays.asList(getTopic(), idempotentTopic));
  }

}
